package com.simon.spark.iplocation

import org.apache.spark.{SparkConf, SparkContext}

object IpLocation {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("iplocation").setMaster("local[2]")
    val sc=new SparkContext(conf);
   val ipLocation= sc.textFile("E:\\BaiduNetdiskDownload\\课程代码\\内存部分\\spark\\day03\\ip.txt")
    .map(x=>(x.split("\\|"))).map(y=>((y(2),y(3),(y(4)+"-"+y(5)+"-"+y(6)+"-"+y(7)+"-"+y(9)),y(13),y(14))))
    val iprulesArray=ipLocation.collect()
    val broadcaseValue=sc.broadcast(iprulesArray)
    //加载要处理的数据
    val shuju=sc.textFile("E:\\BaiduNetdiskDownload\\课程代码\\内存部分\\spark\\day03\\20090121000132.394251.http.format").map(x=>x.split("\\|"))
      .map(y=>y(1)).map(ip=>{
       val ipLong=ipToLong(ip)
       val index=binarySearch(ipLong,broadcaseValue.value);
       val info=broadcaseValue.value(index);
       ((info._4,info._5),1)
    }).reduceByKey(_+_).collect()
    shuju.foreach(println)
  }


  //将ip地址转为对应的long
  def ipToLong(ip:String):Long={
    val ips=ip.split("[.]")
    var ipNum=0L;
    ips.foreach(e=>{
      ipNum=e.toLong | ipNum<<8L
    }
    )
    ipNum
  }
  def binarySearch(ipLong:Long,ips:Array[(String,String,String,String,String)]): Int ={
    var low=0
    var high=ips.length-1
    while(low<=high){
      val middle=(low+high)/2
      if(ipLong>=ips(middle)._1.toLong && ipLong<=ips(middle)._2.toLong){
        return middle
      }
      if(ipLong<ips(middle)._1.toLong){
        high=middle-1
      }else{
        low=middle+1
      }
    }
     -1
  }
}
